package com.jacky.ws.redis;

import com.corundumstudio.socketio.store.RedissonPubSubStore;
import com.corundumstudio.socketio.store.pubsub.PubSubListener;
import com.corundumstudio.socketio.store.pubsub.PubSubMessage;
import com.corundumstudio.socketio.store.pubsub.PubSubType;
import io.netty.util.internal.PlatformDependent;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.codec.SerializationCodec;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

public class CustomRedissonPubSubStore extends RedissonPubSubStore {
    private final RedissonClient redissonPub;
    private final RedissonClient redissonSub;

    private final Long nodeId;
    private final ConcurrentMap<String, Queue<Integer>> map = PlatformDependent.newConcurrentHashMap();

    public CustomRedissonPubSubStore(RedissonClient redissonPub, RedissonClient redissonSub, Long nodeId) {
        super(redissonPub, redissonSub, nodeId);
        this.redissonPub = redissonPub;
        this.redissonSub = redissonSub;
        this.nodeId = nodeId;
    }

    @Override
    public void publish(PubSubType type, PubSubMessage msg) {
        msg.setNodeId(this.nodeId);
        this.redissonPub.getTopic(type.toString(), new SerializationCodec()).publish(msg);
    }

    @Override
    public <T extends PubSubMessage> void subscribe(PubSubType type, PubSubListener<T> listener, Class<T> clazz) {
        String name = type.toString();
        RTopic topic = this.redissonSub.getTopic(name, new SerializationCodec());
        int regId = topic.addListener(clazz, (channel, msg) -> {
            // todo 这里可以根据具体业务场景进行修改
//            if (this.nodeId.equals(msg.getNodeId())) {
            listener.onMessage(msg);
//            }
        });
        Queue<Integer> queue = this.map.get(name);
        if (queue == null) {
            queue = new ConcurrentLinkedQueue<>();
            Queue<Integer> eldQueue = this.map.putIfAbsent(name, queue);
            if (eldQueue != null) {
                queue = eldQueue;
            }
        }
        queue.add(regId);
    }

    @Override
    public void unsubscribe(PubSubType type) {
        String name = type.toString();
        Queue<Integer> regIds = this.map.remove(name);
        RTopic topic = this.redissonSub.getTopic(name, new SerializationCodec());
        for (Integer regId : regIds) {
            topic.removeListener(regId);
        }
    }
}
